package ex55.mm.flow.impound.quote;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import ex55.mm.flow.impound.quote.aggregate.*;
import ex55.mm.flow.impound.quote.common.kit.DateTimeKit;
import ex55.mm.flow.impound.quote.common.rdsdb.RdsDBHelper;
import ex55.mm.flow.impound.quote.exchange.api.MarketDataDepth;
import ex55.mm.flow.impound.quote.exchange.api.Product;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.dbutils.handlers.MapListHandler;
import org.joda.time.DateTime;
import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class AggregateDataCalculator {
    private final Logger logger = LoggerFactory.getLogger(AggregateDataCalculator.class);

    private final AggregateDataFacade aggregateDataFacade;

    private final QueryRunner runner;

    public AggregateDataCalculator(RdsDBHelper rdsDBHelper, AggregateDataFacade aggregateDataFacade) {
        this.runner = new QueryRunner(rdsDBHelper.getDataSourceMaster());
        this.aggregateDataFacade = aggregateDataFacade;
    }

    private Map<String, OnePathRst> cacheBidOnePathRsts = Maps.newConcurrentMap();

    private Map<String, OnePathRst> cacheAskOnePathRsts = Maps.newConcurrentMap();

    private Map<String, Map<String, String>> symbols = Maps.newConcurrentMap();

    private ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();

    public Map<String, OnePathRst> getOnePathRst(String baseCurrency, String quoteCurrency, String pathId) {
        Map<String, OnePathRst> onePathRstMap = Maps.newHashMap();
        try {
            readWriteLock.readLock().lock();
            onePathRstMap.put("bid", cacheBidOnePathRsts.get(quoteCurrency + "$" + pathId));
            onePathRstMap.put("ask", cacheAskOnePathRsts.get(quoteCurrency + "$" + pathId));
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            readWriteLock.readLock().unlock();
        }
        return onePathRstMap;
    }

    public Collection<Map<String, String>> getSymbols() {
        return symbols.values();
    }

    /**
     * 法币兑换需求是AUD =》USD，那么对于USDAUD交易对来说，执行买操作，所以需要看的是卖一盘，
     * 法币兑换算法中AUD =》USD 就生成的是卖盘
     *
     * @param baseCurrency 基础资产
     * @param quoteCurrency 计价资产
     * @return 盘口
     */
    public ConvertDepth getMarketDataDepth(String baseCurrency, String quoteCurrency) {
        ConvertDepth marketDataDepth =
                new ConvertDepth()
                        .setProduct(new Product().setBaseAsset(baseCurrency).setQuoteAsset(quoteCurrency))
                        .setUpdatedAt(DateTime.now().toDate());

        try {
            readWriteLock.readLock().lock();
            TreeMap<BigDecimal, ConvertDepth.Entry> bidsTreeMap = Maps.newTreeMap(Collections.reverseOrder());
            for (OnePathRst onePathRst : cacheBidOnePathRsts.values()) {
                if (onePathRst.availableTradeAmount != null
                        && onePathRst.baseCurrency.equals(baseCurrency)
                        && onePathRst.quoteCurrency.equals(quoteCurrency)) {
                    ConvertDepth.Entry entry = new ConvertDepth.Entry()
                            .setPrice(onePathRst.convertRate)
                            .setQty(onePathRst.availableTradeAmount)
                            .setPathIds(Lists.newArrayList(onePathRst.pathId));
                    bidsTreeMap.merge(onePathRst.convertRate, entry, (entry1, entry2) -> {
                        entry1.setQty(entry1.getQty().add(entry2.getQty()));
                        entry1.getPathIds().addAll(entry2.getPathIds());
                        return entry1;
                    });
                }
            }
            List<ConvertDepth.Entry> bids = Lists.newArrayList();
            bidsTreeMap.forEach((price, entry) -> bids.add(entry));
            marketDataDepth.setBids(bids);

            TreeMap<BigDecimal, ConvertDepth.Entry> asksTreeMap = Maps.newTreeMap();
            for (OnePathRst onePathRst : cacheAskOnePathRsts.values()) {
                if (onePathRst.availableTradeAmount != null
                        && onePathRst.quoteCurrency.equals(quoteCurrency)
                        && onePathRst.baseCurrency.equals(baseCurrency)) {
                    ConvertDepth.Entry entry = new ConvertDepth.Entry()
                            .setPrice(onePathRst.convertRate)
                            .setQty(onePathRst.availableTradeAmount)
                            .setPathIds(Lists.newArrayList(onePathRst.pathId));
                    asksTreeMap.merge(onePathRst.convertRate, entry, (entry1, entry2) -> {
                        entry1.setQty(entry1.getQty().add(entry2.getQty()));
                        entry1.getPathIds().addAll(entry2.getPathIds());
                        return entry1;
                    });
                }
            }

            List<ConvertDepth.Entry> asks = Lists.newArrayList();
            asksTreeMap.forEach((price, entry) -> asks.add(entry));
            marketDataDepth.setAsks(asks);
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            readWriteLock.readLock().unlock();
        }
        return marketDataDepth;
    }

    public SimpleConvertDepth getSimpleMarketDataDepth(String baseCurrency, String quoteCurrency) {
        SimpleConvertDepth marketDataDepth =
                new SimpleConvertDepth()
                        .setProduct(new Product().setBaseAsset(baseCurrency).setQuoteAsset(quoteCurrency))
                        .setUpdatedAt(DateTime.now().toDate());

        try {
            readWriteLock.readLock().lock();
            TreeMap<BigDecimal, SimpleConvertDepth.Entry> bidsTreeMap = Maps.newTreeMap(Collections.reverseOrder());
            for (OnePathRst onePathRst : cacheBidOnePathRsts.values()) {
                if (onePathRst.availableTradeAmount != null
                        && onePathRst.baseCurrency.equals(baseCurrency)
                        && onePathRst.quoteCurrency.equals(quoteCurrency)) {
                    SimpleConvertDepth.Entry entry = new SimpleConvertDepth.Entry()
                            .setPrice(onePathRst.convertRate)
                            .setQty(onePathRst.availableTradeAmount);
                    bidsTreeMap.merge(onePathRst.convertRate, entry, (entry1, entry2) -> {
                        entry1.setQty(entry1.getQty().add(entry2.getQty()));
                        return entry1;
                    });
                }
            }
            List<SimpleConvertDepth.Entry> bids = Lists.newArrayList();
            bidsTreeMap.forEach((price, entry) -> bids.add(entry));
            marketDataDepth.setBids(bids);

            TreeMap<BigDecimal, SimpleConvertDepth.Entry> asksTreeMap = Maps.newTreeMap();
            for (OnePathRst onePathRst : cacheAskOnePathRsts.values()) {
                if (onePathRst.availableTradeAmount != null
                        && onePathRst.quoteCurrency.equals(quoteCurrency)
                        && onePathRst.baseCurrency.equals(baseCurrency)) {
                    SimpleConvertDepth.Entry entry = new SimpleConvertDepth.Entry()
                            .setPrice(onePathRst.convertRate)
                            .setQty(onePathRst.availableTradeAmount);
                    asksTreeMap.merge(onePathRst.convertRate, entry, (entry1, entry2) -> {
                        entry1.setQty(entry1.getQty().add(entry2.getQty()));
                        return entry1;
                    });
                }
            }

            List<SimpleConvertDepth.Entry> asks = Lists.newArrayList();
            asksTreeMap.forEach((price, entry) -> asks.add(entry));
            marketDataDepth.setAsks(asks);
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            readWriteLock.readLock().unlock();
        }
        return marketDataDepth;
    }

    public void start() throws SQLException {
        Set<DBQuotePath> dbQuotePaths = Sets.newHashSet();

        for (Map<String, Object> r : runner.query("select * from mm_impound_quote_path;", new MapListHandler())) {
            String path_content = r.get("path_content").toString();
            dbQuotePaths.add(
                    new DBQuotePath()
                            .setSource(r.get("source").toString())
                            .setPathId(r.get("path_id").toString())
                            .setTarget(r.get("target").toString())
                            .setExEdges(JSON.parseArray(path_content, ExEdge.class)));
        }

        Runnable runnable =
                () -> {
                    while (true) {
                        try {
                            Thread.sleep(10 * 1000);

                            Map<String, OnePathRst> bidOnePathRsts = Maps.newConcurrentMap();
                            Map<String, OnePathRst> askOnePathRsts = Maps.newConcurrentMap();

                            Map<String, MarketDataDepth> dataDepth =
                                    aggregateDataFacade.getData_MarketDataDepth();
                            Map<String, ExchangeProduct> exchangeProducts =
                                    aggregateDataFacade.getData_ExchangeProduct();

                            for (DBQuotePath dbQuotePath : dbQuotePaths) {
                                try {
                                    Map<String, String> symbolEntry = Maps.newHashMap();
                                    symbolEntry.put("baseCurrency", dbQuotePath.getSource());
                                    symbolEntry.put("quoteCurrency", dbQuotePath.getTarget());
                                    symbols.putIfAbsent(dbQuotePath.getSource() + dbQuotePath.getTarget(), symbolEntry);
                                    List<OnePathRst> _bidOnePathRsts =
                                            checkOnePaths(
                                                    dbQuotePath.getPathId(),
                                                    dbQuotePath.getSource(),
                                                    dbQuotePath.getTarget(),
                                                    dbQuotePath.getExEdges(),
                                                    exchangeProducts,
                                                    dataDepth);

                                    _bidOnePathRsts.forEach(bidOnePathRst -> {
                                        // 处理统计结果
                                        if (bidOnePathRst.pathSuccess&& bidOnePathRst.convertRate != null) {
                                            assert bidOnePathRst.availableTradeAmount != null;
                                            bidOnePathRst.availableTradeAmount = bidOnePathRst.availableTradeAmount
                                                    .divide(bidOnePathRst.convertRate, 12, RoundingMode.DOWN).stripTrailingZeros();
                                            // 生成的是买盘，价格应该更低
                                            bidOnePathRst.convertRate = bidOnePathRst.convertRate
                                                    .multiply(BigDecimal.ONE.subtract(bidOnePathRst.costRate).subtract(bidOnePathRst.tradeWeight)).setScale(12, RoundingMode.DOWN).stripTrailingZeros();
                                            bidOnePathRst.source = dbQuotePath.getSource();
                                            bidOnePathRst.target = dbQuotePath.getTarget();
                                            logger.info("onePathRst success:{}", JSON.toJSONString(bidOnePathRst));
                                            bidOnePathRsts.put( dbQuotePath.getTarget() + "$" + bidOnePathRst.pathId, bidOnePathRst);
                                        } else {
                                            logger.warn(
                                                    "onePathRst fail:{}, {}",
                                                    bidOnePathRst.breakKey,
                                                    bidOnePathRst.breakMessage);
                                        }
                                    });

                                    List<OnePathRst> _askOnePathRsts =
                                            checkOnePaths(
                                                    dbQuotePath.getPathId(),
                                                    dbQuotePath.getTarget(),
                                                    dbQuotePath.getSource(),
                                                    dbQuotePath.getExEdges(),
                                                    exchangeProducts,
                                                    dataDepth);
                                    _askOnePathRsts.forEach(askOnePathRst -> {
                                        // 处理统计结果
                                        if (askOnePathRst.pathSuccess && askOnePathRst.convertRate != null) {
                                            askOnePathRst.baseCurrency = dbQuotePath.getSource();
                                            askOnePathRst.quoteCurrency = dbQuotePath.getTarget();
                                            // 生成的是卖盘，价格应该更高
                                            askOnePathRst.convertRate = BigDecimal.ONE
                                                    .divide(askOnePathRst.convertRate, 12, RoundingMode.DOWN)
                                                    .multiply(BigDecimal.ONE.add(askOnePathRst.costRate).add(askOnePathRst.tradeWeight)).setScale(12, RoundingMode.DOWN);
                                            askOnePathRst.source = dbQuotePath.getTarget();
                                            askOnePathRst.target = dbQuotePath.getSource();
                                            logger.info("onePathRst success:{}", JSON.toJSONString(askOnePathRst));
                                            askOnePathRsts.put(dbQuotePath.getTarget() + "$" + askOnePathRst.pathId, askOnePathRst);
                                        } else {
                                            logger.warn(
                                                    "onePathRst fail:{}, {}",
                                                    askOnePathRst.breakKey,
                                                    askOnePathRst.breakMessage);
                                        }
                                    });
                                } catch (Exception e) {
                                    e.printStackTrace();
                                }
                            }

                            if (bidOnePathRsts.size() > 0 || askOnePathRsts.size() > 0) {
                                updateCache(bidOnePathRsts, askOnePathRsts);
                            }

                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                };
        new Thread(runnable).start();
    }

    private void updateCache(Map<String, OnePathRst> bidOnePathRsts, Map<String, OnePathRst> askOnePathRsts) {
        try {
            readWriteLock.writeLock().lock();
            cacheAskOnePathRsts.clear();
            cacheBidOnePathRsts.clear();

            cacheAskOnePathRsts.putAll(askOnePathRsts);
            cacheBidOnePathRsts.putAll(bidOnePathRsts);
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally{
            readWriteLock.writeLock().unlock();
        }
    }

    private <N> N index(List<N> list, int index) {
        if (list == null) {
            throw new IllegalArgumentException("list is null");
        }
        if (list.size() >= index) {
            return list.get(index);
        } else {
            throw new IllegalArgumentException("index out");
        }
    }

    private static List<ExEdge> subExEdge(String baseCurrency, String quoteCurrency, List<ExEdge> exEdges) {
        List<ExEdge> newExEdges = Lists.newArrayList();

        int startIndex = 0;
        for (int i = 0; i < exEdges.size(); i++) {
            ExEdge exEdge = exEdges.get(i);
            if (exEdge.getSource().endsWith(baseCurrency) && !exEdge.getType().equals(ExEdge.Type.TRANSFER)) {
                startIndex = i;
                break;
            }
        }

        newExEdges.addAll(exEdges.subList(startIndex, exEdges.size()));
        newExEdges.addAll(exEdges.subList(0, startIndex));

        List<ExEdge> result = Lists.newArrayList();

        for (ExEdge exEdge : newExEdges) {
            result.add(exEdge);
            if (exEdge.getTarget().endsWith(quoteCurrency)) {
                break;
            }
        }
        return result;
    }

    private List<OnePathRst> checkOnePaths(
            String pathId,
            String baseCurrency,
            String quoteCurrency,
            List<ExEdge> onePath,
            Map<String, ExchangeProduct> exchangeProducts,
            Map<String, MarketDataDepth> depths) {

        List<ExEdge> realPath = subExEdge(baseCurrency, quoteCurrency, onePath);

        OnePathRst standardOnePathRst = new OnePathRst();
        standardOnePathRst.path = realPath;
        standardOnePathRst.pathId = pathId;
        standardOnePathRst.baseCurrency = baseCurrency;
        standardOnePathRst.quoteCurrency = quoteCurrency;
        List<List<Integer>> depthSteps = Lists.newArrayList();

        for (ExEdge edge : realPath) {
            ExEdge.Type type = edge.getType();
            if (type.equals(ExEdge.Type.TRANSFER)) {
                standardOnePathRst.costRate = standardOnePathRst.costRate.add(BigDecimal.ZERO);
            } else {
                standardOnePathRst.tradeCount++;
                String label = edge.getLabel();
                String productId = edge.getProductId();

                TradeEdgeRst tradeEdgeRst;

                // 1. 过滤异常数据
                ExchangeProduct exchangeProduct = exchangeProducts.get(productId);
                if (exchangeProduct == null) {
                    // 不通了
                    standardOnePathRst.breakPath(
                            label,
                            "Break by :"
                                    + label
                                    + " , Message:exchangeProduct == null , PathId:"
                                    + pathId);
                    depthSteps.clear();
                    standardOnePathRst.pathSuccess = false;
                    break;
                }

                BigDecimal rawTradeTakerFee = exchangeProduct.getTradeTakerFee();
                if (rawTradeTakerFee == null) {
                    // 不通了
                    standardOnePathRst.breakPath(
                            label,
                            "Break by :"
                                    + label
                                    + " , Message: rawTradeTakerFee == null , PathId:"
                                    + pathId);
                    depthSteps.clear();
                    standardOnePathRst.pathSuccess = false;
                    break;
                }

                MarketDataDepth depth = depths.get(productId);
                if (depth == null) {
                    String message = "depth is null";
                    // 不通了
                    standardOnePathRst.breakPath(
                            label,
                            "Break by :" + label + " , Message:" + message + " , PathId:" + pathId);
                    depthSteps.clear();
                    standardOnePathRst.pathSuccess = false;
                    break;
                }

                if (depth.getRefreshedAt().getTime()
                        < System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(60 + 30)) {
                    String message =
                            "depth updated at unable "
                                    + DateTimeKit.stringUTCISO8601(depth.getRefreshedAt())
                                    + " now is:"
                                    + DateTimeKit.stringUTCISO8601(standardOnePathRst.created_at)
                                    + ", depth:"
                                    + JSON.toJSONString(depth);
                    // 不通了
                    standardOnePathRst.breakPath(
                            label,
                            "Break by :" + label + " , Message:" + message + " , PathId:" + pathId);
                    depthSteps.clear();
                    standardOnePathRst.pathSuccess = false;
                    break;
                }

                if (CollectionUtils.isEmpty(depth.getBids())
                        || CollectionUtils.isEmpty(depth.getAsks())) {
                    String message = "asks or bids isEmpty";
                    // 不通了
                    standardOnePathRst.breakPath(
                            label,
                            "Break by :" + label + " , Message:" + message + " , PathId:" + pathId);
                    depthSteps.clear();
                    standardOnePathRst.pathSuccess = false;
                    break;
                }

                BigDecimal tradeTakerWeight = exchangeProduct.getWeight();

                if (type.equals(ExEdge.Type.TRADE_BUY)) {
                    tradeEdgeRst =
                            new TradeEdgeRst(
                                    label,
                                    ExEdge.Type.TRADE_BUY,
                                    depth,
                                    exchangeProduct,
                                    tradeTakerWeight,
                                    rawTradeTakerFee);
                } else if (type.equals(ExEdge.Type.TRADE_SELL)) {
                    tradeEdgeRst =
                            new TradeEdgeRst(
                                    label,
                                    ExEdge.Type.TRADE_SELL,
                                    depth,
                                    exchangeProduct,
                                    tradeTakerWeight,
                                    rawTradeTakerFee);
                } else {
                    standardOnePathRst.breakPath(
                            label,
                            "Break by :"
                                    + label
                                    + " , Message: type error,"
                                    + type
                                    + " , Path:"
                                    + realPath);
                    depthSteps.clear();
                    standardOnePathRst.pathSuccess = false;
                    break;
                }

                if (type.equals(ExEdge.Type.TRADE_SELL)) {
                    // 2. 盘口深度
                    List<Integer> depthStep = Lists.newArrayList();
                    for (int i=0; i< depth.getBids().size(); i++) {
                        depthStep.add(i);
                    }
                    depthSteps.add(depthStep);
                } else {
                    List<Integer> depthStep = Lists.newArrayList();
                    for (int i=0; i< depth.getAsks().size(); i++) {
                        depthStep.add(i);
                    }
                    depthSteps.add(depthStep);
                }

                standardOnePathRst.tradeEdge.add(tradeEdgeRst);
            }
        }

        List<OnePathRst> onePathRsts = Lists.newArrayList();

        if (CollectionUtils.isEmpty(depthSteps)) {
            onePathRsts.add(standardOnePathRst);
            return onePathRsts;
        }

        List<List<Integer>> decare = Lists.cartesianProduct(depthSteps);

        decare.forEach(steps -> {
            // 每个组合[0,0,0]=> step
            OnePathRst newOnePathRst = new OnePathRst();
            copyProperties(standardOnePathRst, newOnePathRst);
            newOnePathRst.costRate = BigDecimal.ZERO;
            newOnePathRst.tradeWeight = BigDecimal.ZERO;

            for (int i = 0; i < newOnePathRst.tradeEdge.size(); i++) {
                TradeEdgeRst tradeEdgeRst = newOnePathRst.tradeEdge.get(i);

                if (tradeEdgeRst.side.equals(ExEdge.Type.TRADE_SELL)) {
                    MarketDataDepth.Entry bid = index(tradeEdgeRst.depth.getBids(), steps.get(i));
                    if (newOnePathRst.convertRate == null) {
                        newOnePathRst.convertRate =
                                BigDecimal.ONE.multiply(bid.getPrice());
                    } else {
                        newOnePathRst.convertRate =
                                newOnePathRst.convertRate.multiply(bid.getPrice());
                    }

                    if (newOnePathRst.availableTradeAmount == null) {
                        newOnePathRst.availableTradeAmount = bid.getQuantity();
                    } else {
                        BigDecimal qty = bid.getQuantity();
                        if (newOnePathRst.availableTradeAmount.compareTo(qty) > 0) {
                            newOnePathRst.availableTradeAmount = qty;
                        }
                        newOnePathRst.availableTradeAmount = newOnePathRst.availableTradeAmount.multiply(bid.getPrice());
                    }
                } else {
                    MarketDataDepth.Entry ask = index(tradeEdgeRst.depth.getAsks(), steps.get(i));

                    if (newOnePathRst.convertRate == null) {
                        newOnePathRst.convertRate =
                                BigDecimal.ONE.divide(
                                        ask.getPrice(),
                                        15,
                                        RoundingMode.DOWN);
                    } else {
                        newOnePathRst.convertRate =
                                newOnePathRst.convertRate.divide(
                                        ask.getPrice(),
                                        15,
                                        RoundingMode.DOWN);
                    }

                    if (newOnePathRst.availableTradeAmount == null) {
                        newOnePathRst.availableTradeAmount = ask.getQuantity();
                    } else {
                        BigDecimal amount = ask.getPrice().multiply(ask.getQuantity());
                        if (newOnePathRst.availableTradeAmount.compareTo(amount) > 0) {
                            newOnePathRst.availableTradeAmount = amount;
                        }
                        newOnePathRst.availableTradeAmount = newOnePathRst.availableTradeAmount.divide(ask.getPrice(), 15, RoundingMode.DOWN);
                    }
                }

                if (i == 0) {
                    newOnePathRst.pathId = newOnePathRst.pathId + "$" + steps.get(i);
                } else {
                    newOnePathRst.pathId = newOnePathRst.pathId + "_" + steps.get(i);
                }

                // 计算当前边
                newOnePathRst.costRate =
                        newOnePathRst.costRate
                                .add(tradeEdgeRst.rawTradeTakerFee);

                newOnePathRst.tradeWeight = newOnePathRst.tradeWeight.add(tradeEdgeRst.tradeWeight);
            }

            onePathRsts.add(newOnePathRst);
        });
        return onePathRsts;
    }

    private void copyProperties(OnePathRst source, OnePathRst target) {
        if (source == null || target == null) {
            throw new IllegalArgumentException("source or target is null");
        }

        target.tradeEdge = source.tradeEdge;
        target.pathId = source.pathId;
        target.path = source.path;
        target.tradeCount = source.tradeCount;
        target.baseCurrency = source.baseCurrency;
        target.quoteCurrency = source.quoteCurrency;
        target.pathSuccess = source.pathSuccess;
    }
}
